#include <rise/common/Log.h>
#include <rise/common/exmacros.h>
#include <rise/common/ExceptionTemplate.h>
#include <rise/threading/Thread.h>
#include <staff/common/Value.h>
#include <staff/common/Operation.h>
#include <staff/common/DataObject.h>
#include "Client.h"
#include "Message.h"
#include "Subscriber.h"

namespace smq
{
  struct Subscriber::SubscriberImpl: public rise::threading::CThread
  {
    Subscriber* pSelf;
    std::string sSubscriberName;
    MessageHandler* pMessageHandler;
    Client tClient;
    bool bThread;

    SubscriberImpl(Subscriber* pSelfInit):
      pSelf(pSelfInit), pMessageHandler(NULL), bThread(false)
    {
    }

    void Invoke(staff::COperation& rOperation)
    {
      tClient.Post(rOperation.Request());
      std::string sResultNodeName = rOperation.Request().GetLocalName() + "Result";
      staff::CDataObject tResponse;
      for(;;)
      {
        tClient.Receive(tResponse, false);
        const std::string& sReceivedNodeName = tResponse.GetLocalName();
        if (sReceivedNodeName == sResultNodeName)
        {
          break;
        }
        else
        if (sReceivedNodeName == "Fault")
        {
          RISE_THROWS(rise::CLogicFormatException, tResponse.GetText());
        }
        else
        {
          rise::LogWarning() << "Unknown server message: " << sReceivedNodeName;
        }
      }
      rOperation.GetResponse() = tResponse;
    }
    
    void Run(void*)
    {
      SubscriberWorker();
    }
    
    void SubscriberWorker()
    {
      for(;;)
      {
        Message tMsg;
        ReceiveMessage(tMsg, false);
        if (tClient.IsShuttingDown())
        {
          break;
        }

        if (pMessageHandler)
        {
          (*(pMessageHandler))(*pSelf, tMsg);
        }
      }
    }

    void ReceiveMessage(Message& rMsg, bool bRemove)
    {
      staff::CDataObject tdoMessage;
  
      tClient.Receive(tdoMessage, true);
  
      if (tClient.IsShuttingDown())
      {
        return;
      }

      tdoMessage >> rMsg;
  
      if (bRemove)
      {
        RemoveMessage(rMsg.sQueue, rMsg.sId);
      }
    }

    void RemoveMessage(const std::string& sQueueName, const std::string& sMsgId)
    {
      staff::COperation tOperation("GetQueueLength");
      tOperation.AddParameter("sQueueName", sQueueName);
      tOperation.AddParameter("sMsgId", sMsgId);
      Invoke(tOperation);
    }

  };


  Subscriber::Subscriber()
  {
    m_pImpl = new SubscriberImpl(this);
  }

  Subscriber::~Subscriber()
  {
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  void Subscriber::Init(const std::string& sSubscriberName, const std::string& sServerHost /*= "127.0.0.1"*/, unsigned short ushServerPort /*= 8768*/)
  {
    m_pImpl->tClient.Init(sServerHost, ushServerPort);
    m_pImpl->sSubscriberName = sSubscriberName;

    staff::COperation tOperation("Ident");
    tOperation.AddParameter("sName", sSubscriberName);
    tOperation.AddParameter("sType", "subscriber");
    m_pImpl->Invoke(tOperation);
  }

  void Subscriber::SetHandler(MessageHandler* pMessageHandler)
  {
    m_pImpl->pMessageHandler = pMessageHandler;
  }

  void Subscriber::Subscribe(const std::string& sQueueName)
  {
    staff::COperation tOperation("Subscribe");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
  }

#ifndef STAFFMQ_NO_DEPRECATED
  void Subscriber::Subscribe(const std::string& sQueueName, MessageHandler* pMessageHandler)
  {
    rise::LogWarning() << "This function is deprecated. Consider using Subscriber::SetHandler and Subscriber::Subscribe";
    SetHandler(pMessageHandler);
    Subscribe(sQueueName);
  }
#endif //  STAFFMQ_NO_DEPRECATED

  void Subscriber::Unsubscribe(const std::string& sQueueName)
  {
    staff::COperation tOperation("Unsubscribe");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
  }

  void Subscriber::Start(bool bThread /*= false*/)
  {
    m_pImpl->bThread = bThread;
    if (bThread)
    {
      m_pImpl->Start(NULL, false);
    }
    else
    {
      m_pImpl->SubscriberWorker();
    }
  }

  void Subscriber::Stop()
  {
    m_pImpl->tClient.Shutdown();
    if (m_pImpl->bThread)
    {
      m_pImpl->JoinThread();
    }
    m_pImpl->tClient.Close();
  }

  bool Subscriber::IsStopping()
  {
    return m_pImpl->tClient.IsShuttingDown();
  }

  bool Subscriber::IsWorking() const
  {
    return m_pImpl->IsWorking();
  }
    
  void Subscriber::Wait()
  {
    if (m_pImpl->bThread)
    {
      m_pImpl->JoinThread();
    }
  }

  void Subscriber::GetQueueList(StringList& rlsQueues)
  {
    staff::COperation tOperation("GetQueueList");
    m_pImpl->Invoke(tOperation);

    staff::CDataObject& rdoList = tOperation.Result();
    rlsQueues.clear();
    for (staff::CDataObject::Iterator itHeader = rdoList.Begin();
      itHeader != rdoList.End(); ++itHeader)
    {
      rlsQueues.push_back(itHeader->GetText());
    }
  }

  void Subscriber::GetFilteredQueueList(const std::string& sFilter, StringList& rlsQueues)
  {
    staff::COperation tOperation("GetFilteredQueueList");
    tOperation.AddParameter("sFilter", sFilter);
    m_pImpl->Invoke(tOperation);

    staff::CDataObject& rdoList = tOperation.Result();
    rlsQueues.clear();
    for (staff::CDataObject::Iterator itHeader = rdoList.Begin();
      itHeader != rdoList.End(); ++itHeader)
    {
      rlsQueues.push_back(itHeader->GetText());
    }
  }

  int Subscriber::GetQueueLength(const std::string& sQueueName)
  {
    staff::COperation tOperation("GetQueueLength");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
    const staff::CValue& rValue = tOperation.ResultValue(); // gcc 2.9x warning
    return rValue;
  }

  void Subscriber::GetLastMessage(const std::string& sQueueName, Message& rMsg)
  {
    staff::COperation tOperation("GetLastMessage");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
    staff::CDataObject tdoMessage = tOperation.Result().GetChildByLocalName("Message");
    tdoMessage >> rMsg;
  }

  void Subscriber::GetMessageHeadersList(const std::string& sQueueName, MessageHeaderList& rlsMessages)
  {
    staff::COperation tOperation("GetMessageHeadersList");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);

    staff::CDataObject& rdoList = tOperation.Result();
    rlsMessages.clear();
    for (staff::CDataObject::Iterator itHeader = rdoList.Begin();
      itHeader != rdoList.End(); ++itHeader)
    {
      MessageHeader stHeader;
      staff::CDataObject rdoHeader = *itHeader;

      rdoHeader >> stHeader;
      rlsMessages.push_back(stHeader);
    }
  }

  void Subscriber::GetMessage(const std::string& sQueueName, const std::string& sId, Message& rMsg)
  {
    staff::COperation tOperation("GetMessage");
    tOperation.AddParameter("sQueueName", sQueueName);
    tOperation.AddParameter("sId", sId);
    m_pImpl->Invoke(tOperation);
    staff::CDataObject tdoMessage = tOperation.Result().GetChildByLocalName("Message");
    tdoMessage >> rMsg;
  }


  void Subscriber::ReceiveMessage(Message& rMsg, bool bRemove /*= false*/)
  {
    m_pImpl->ReceiveMessage(rMsg, bRemove);
  }

  void Subscriber::RemoveMessage(const std::string& sQueueName, const std::string& sMsgId)
  {
    m_pImpl->RemoveMessage(sQueueName, sMsgId);
  }

}
